[core][spark] Report skipped manifests, resulted file size and record count in scan metrics - #9657
[core][spark] Report skipped manifests, resulted file size and record count in scan metrics#9657zhuxiangyi wants to merge 1 commit into
Conversation
… count in scan metrics Scan metrics reported how much was read but not whether that amount was reasonable. lastScannedManifests only reported the count after manifest level filtering, so the pruning ratio could not be computed, and only file counts were reported, so the data volume of a scan could not be estimated. Add lastScanSkippedManifests, lastScanResultedTableFilesSize and lastScanResultedRecordCount, computed at the existing reporting site in AbstractFileStoreScan#plan from values that are already in memory, and expose them as Spark custom metrics.
c9583ca to
83705bb
Compare
JingsongLi
left a comment
There was a problem hiding this comment.
Pruning counts and planned bytes/records have real diagnostic value, but the new ADD-only calculation does not represent all supported read paths. Details are inline. The two ScanMetricsTest cases pass with head classes on JDK 8; they validate gauge registration/reporting but do not exercise the DELTA plan that exposes the issue. I have not rerun the Spark metrics suite.
| long resultedRecordCount = 0L; | ||
| for (ManifestEntry entry : result) { | ||
| if (entry.kind() == FileKind.ADD) { | ||
| resultedTableFilesSize += entry.file().fileSize(); |
There was a problem hiding this comment.
[P2] Include before-files when reporting the cost of change reads
The assumption that DELETE entries are never read is false for SnapshotReaderImpl.readChanges(): it uses this DELTA plan's DELETE entries as IncrementalSplit.beforeFiles, and IncrementalDiffSplitRead/IncrementalChangelogReadProvider actually read those files to produce changes. FollowUpScanner reaches this path for overwrite changes, and AllDeltaFollowUpScanner uses it for delta reads. A deletion-only change can therefore report zero resulted bytes/records while scanning the old files; a replacement underreports the before side.
Keep the metrics' documented meaning tied to the files the consumer will actually read, for example by calculating them after the normal-read versus change-read split selection. Do not unconditionally sum all DELETE entries for ordinary ADD-only reads. Add a DELTA/overwrite-changelog regression alongside the existing batch gauge tests.
Purpose
Scan metrics today tell the user how much was read, but not whether that amount is
reasonable. Two concrete gaps:
1. Manifest pruning efficiency cannot be computed — the numerator is reported without the denominator.
lastScannedManifestsreportsmanifestsResult.filteredManifests.size(), i.e. the countafter manifest level filtering. The total (
allManifests.size()) sits on the very sameobject but is never reported, so
scannedManifests = 200is ambiguous:Case B is a common production problem (a predicate that fails to push down because of a
function or a type mismatch on the partition column), and it is exactly the case the current
metrics cannot distinguish.
2. Only file counts are reported, no bytes and no records.
resultedTableFiles = 3000does not answer "how much data will this query read". In Paimonthe relation between file count and data volume is unstable: a frequently written table may
have 3000 files holding 2 GB, while the same table after compaction may have 300 files
holding 20 GB. Deciding whether a query is slow because of data volume, or whether a table
needs compaction, requires bytes and records. Both
DataFileMeta#fileSizeandDataFileMeta#rowCountare already carried by the entries in the scan result.There is also an asymmetry with the write side:
CommitStatsalready reports record levelcounters (
deltaRecordsAppended,changelogRecordsAppended), while the scan side reportsfile counts only.
Changes
Three fields are added to
ScanStatsand three gauges toScanMetrics:lastScanSkippedManifestsallManifests.size() - filteredManifests.size()lastScanResultedTableFilesSizelastScanResultedRecordCountTogether with the existing metrics this completes two axes that are currently incomplete:
scannedManifests,skippedManifests,skippedTableFilesresultedTableFiles,resultedTableFilesSize,resultedRecordCountThey are computed at the single existing reporting site in
AbstractFileStoreScan#plan,inside the existing
if (scanMetrics != null)block, so there is no cost when metrics aredisabled:
Note the deliberate asymmetry: size and record count only cover
FileKind.ADDentries,because for
DELTA/CHANGELOGscan modes the result also carriesDELETEentries whosebytes will never be read. The existing
resultedTableFileskeeps its current semantics (allkinds), so no existing metric changes value.
The three metrics are also exposed as Spark custom metrics (
PaimonMetrics,SparkMetricRegistry,PaimonBaseScan#supportedCustomMetrics); the size one usesPaimonSizeSumMetricso the SQL tab renders18.2 GiBrather than a raw byte count. TheFlink side needs no change,
FlinkMetricRegistryforwards the new gauges automatically.This PR also fixes a typo in the
planningDurationmetric description (planing->planning), which is in the same block of code.Compatibility
No format change. Both source values are already persisted fields
(
ManifestFileMeta#_NUM_ADDED_FILES,DataFileMeta#_FILE_SIZE); nothing new is written andthe write path is never entered, so old tables, old readers and rolling Flink upgrades are
unaffected. The new gauges are purely additive — no existing metric name or semantic changes,
so existing dashboards keep working.
For reference, Iceberg reports the equivalent counters in its
ScanMetrics:TOTAL_DATA_MANIFESTS,SKIPPED_DATA_MANIFESTSandTOTAL_FILE_SIZE_IN_BYTES.Follow-ups (not in this PR)
scanDurationdown into manifest IO vs. filtering, so a slow plan can be attributed.resultedLevel0Files, to show how much un-compacted data a query has to read.Tests
ScanMetricsTest— extended to cover the three new gauges (registration, initial values andvalues after each report).
PaimonMetricTest—checkMetricsnow also assertsresultedRecordCountand thatresultedTableFilesSize > 0, plus a new assertion that a scan without any filter cannotprune any manifest (
skippedManifests == 0). Passes under both-Pspark3(Scala 2.12) and-Pspark4(Scala 2.13).FileStoreSourceMetricsTest— unchanged and still passing, confirming the Flink side needsno change.